import pika
import json
import uuid
from threading import Lock
import multiprocessing
from . import mqRPC_exception, mqRPC_deadMsg_consumer
from .load_config import config

# ### 说明 ###
# 该client为线程安全的
# 每个client 有两个回调队列（一个用于同步RPC,一个异步RPC）


class MqRPCClient:
    # 这个MqRPCClient作为父类

    __dead_msg_consumer_process = None

    # rpc server和死信的队列名 这在自动生成的代码里设置
    server_queue = None
    dead_queue = None

    def __init__(self, async=False, local_dead_msg_consumer=True):
        # 是否为异步RPC
        self.__async = async
        # 对紧接着的下一条RPC求设置该超时时间 请求完了又设置为None不超时
        self.signal_request_timeout = None

        # 异步回调队列名
        self.__async_callback_queue = None
        # 同步回调队列名
        self.__sync_callback_queue = None

        # python内置dict list为线程安全 不用设置互斥量
        # # 两个互斥量
        # self.__async_mutex = Lock()
        # self.__sync_mutex = Lock()

        # 异步或同步 process_data_events 互斥量
        self.__sync_process_mutex = Lock()
        self.__async_process_mutex = Lock()

        # 存储异步RPC消息的必要数据
        # 格式 {corr_id:{response:,func_name:}}
        # 当通过get_response()从队列中取出结果后 从其中删除该条请求的数据
        # 对他的操作应该是互斥的
        self.__async_responses = {}

        # 暂存同步RPC消息结果
        # 这里为了线程安全由原来的暂存变量变为由corr_id标识的字典
        # 格式 {corr_id:responses}
        # 对他的操作应该是互斥的
        self.__sync_responses = {}

        # rabbit mq连接凭证 和 参数
        self.credentials = pika.PlainCredentials(config["rabbitmq_user"], config["rabbitmq_pwd"])
        self.connect_paras = pika.ConnectionParameters(host=config["rabbitmq_host"], port=config.getint("rabbitmq_port")
                                                       , credentials=self.credentials,
                                                       virtual_host=config["rabbitmq_vhost"])

        # 连接 mq 声明需要使用到的 queue
        self.connect()

        # 启动本地静态死信consumer
        if local_dead_msg_consumer:
            MqRPCClient.__launch_local_dead_msg_consumer()

    # # 对 __sync_response的互斥操作
    # def __del_sync_responses(self, corr_id):
    #     self.__sync_mutex.acquire()
    #     del self.__sync_responses[corr_id]
    #     self.__sync_mutex.release()
    #
    # def __change_sync_responses(self, corr_id, response):
    #     self.__sync_mutex.acquire()
    #     self.__sync_responses[corr_id] = response
    #     self.__sync_mutex.release()
    #
    # # 对 __async_response的互斥操作
    # def __del_async_responses(self, corr_id):
    #     self.__async_mutex.acquire()
    #     del self.__async_responses[corr_id]
    #     self.__async_mutex.release()
    #
    # def __change_async_responses(self, corr_id, callback_queue, response):
    #     self.__async_mutex.acquire()
    #     self.__async_responses[corr_id] = {"callback_queue": callback_queue, "response": response}
    #     self.__async_mutex.release()

    @property
    def async(self):
        return self.__async

    # @async.setter
    # def async(self, value):
    #     self.__async = value

    # 连接和声明队列交换机
    def connect(self):
        # 连接mq
        self.connection = pika.BlockingConnection(self.connect_paras)
        self.channel = self.connection.channel()

        # 声明RPC队列 独占即连接断开后会被删除
        result = self.channel.queue_declare(exclusive=True)
        self.__async_callback_queue = result.method.queue

        result = self.channel.queue_declare(exclusive=True)
        self.__sync_callback_queue = result.method.queue

        # 设置回调函数
        self.channel.basic_consume(self.on_response, queue=self.__async_callback_queue, no_ack=True)
        self.channel.basic_consume(self.on_response, queue=self.__sync_callback_queue, no_ack=True)

    # 异步发送请求
    def __async_send_request(self, corr_id, body, func_name):
        # 将必要消息存入异步消息列表中
        # self.__change_async_responses(corr_id, self.__async_callback_queue, None)
        self.__async_responses[corr_id] = {"response": None, "func_name": func_name}
        # 发布RPC请求消息
        self.channel.basic_publish(
            exchange='',
            routing_key=MqRPCClient.server_queue,
            properties=pika.BasicProperties(
                reply_to=self.__async_callback_queue,
                correlation_id=corr_id,
                expiration=self.signal_request_timeout
            ),
            body=json.dumps(body))
        self.signal_request_timeout = None
        return corr_id

    # 同步发送请求
    def __sync_send_request(self, corr_id, body, func_name):
        # self.__change_sync_responses(corr_id, None)
        self.__sync_responses[corr_id] = None
        # 发布RPC请求消息
        self.channel.basic_publish(
            exchange='',
            routing_key=MqRPCClient.server_queue,
            properties=pika.BasicProperties(
                reply_to=self.__sync_callback_queue,
                correlation_id=corr_id,
                expiration=self.signal_request_timeout
            ),
            body=json.dumps(body))
        self.signal_request_timeout = None
        # 循环获取
        while True:
            # 这里加锁 保证检查数据后紧跟着消费数据
            self.__sync_process_mutex.acquire()
            if self.__sync_responses[corr_id]:
                self.__sync_process_mutex.release()
                break
            self.connection.process_data_events()
            self.__sync_process_mutex.release()
        # 直接从服务端返回的response
        response = self.__sync_responses[corr_id]
        # 从暂存队列中删除
        # self.__del_sync_responses(corr_id)
        del self.__sync_responses[corr_id]
        if response["status"] != "200":
            # RPC执行异常
            raise mqRPC_exception.MqRPCException.get_exception(response["exception"], response["status"], func_name,
                                                               response["reason"])
        return response["data"]

    def send_request(self, body, func_name):
        if self.connection.is_closed:
            # 连接已经断开则重新连接
            self.connect()
            if not self.connection.is_open:
                # 没有连接上抛出异常
                raise pika.connection.exceptions.AMQPConnectionError()

        # 相关性id 同时作为一个请求独一无二的标识符
        corr_id = str(uuid.uuid4())

        if self.__async:
            return self.__async_send_request(corr_id, body, func_name)
        else:
            return self.__sync_send_request(corr_id, body, func_name)

    # 异步获取请求响应
    def get_response(self, corr_id):
        # 循环获取
        while True:
            self.__async_process_mutex.acquire()
            if self.__async_responses[corr_id]["response"]:
                self.__async_process_mutex.release()
                break
            self.connection.process_data_events()
            self.__async_process_mutex.release()
        # 直接从服务端返回的response
        response = self.__async_responses[corr_id]["response"]
        func_name = self.__async_responses[corr_id]["func_name"]
        # 从暂存队列中删除
        del self.__async_responses[corr_id]
        if response["status"] != "200":
            # RPC执行异常
            raise mqRPC_exception.MqRPCException.get_exception(response["exception"], response["status"], func_name,
                                                               response["reason"])
        return response["data"]

    # 接收到响应 回调函数
    def on_response(self, ch, method, props, body):
        if self.async:
            # 如果这条消息在异步等待结果列表里就设置，否则就丢弃
            if props.correlation_id in self.__async_responses:
                # self.__change_async_responses(props.correlation_id,
                #                               self.__async_responses[props.correlation_id]["callback_queue"],
                #                               json.loads(str(body, encoding='utf8')))
                self.__async_responses[props.correlation_id]["response"] = json.loads(str(body, encoding='utf8'))
        else:
            # 如果这条消息 在暂存同步结果列表里就设置，否则就丢弃
            if props.correlation_id in self.__sync_responses:
                # self.__change_sync_responses(props.correlation_id, json.loads(str(body, encoding='utf8')))
                self.__sync_responses[props.correlation_id] = json.loads(str(body, encoding='utf8'))

    def close_connection(self):
        if not self.connection.is_closed:
            self.connection.close()

    @staticmethod
    def close_local_dead_msg_consumer():
        if MqRPCClient.__dead_msg_consumer_process.is_alive():
            MqRPCClient.__dead_msg_consumer_process.terminate()

    @staticmethod
    def _start_dead_msg_consumer():
        dead_consumer = mqRPC_deadMsg_consumer.MqRPCDeadMsgConsumer(MqRPCClient.dead_queue)
        dead_consumer.start_consume()

    @staticmethod
    def __launch_local_dead_msg_consumer():
        # 本地开启一个进程用作死信处理
        # 建议不使用 这不是一个很好的方法（这是一个过渡的方法）
        # 如果是不同RPC的client会启动相应的本地进程
        if MqRPCClient.__dead_msg_consumer_process and MqRPCClient.__dead_msg_consumer_process.is_alive():
            return
        else:
            MqRPCClient.__dead_msg_consumer_process = multiprocessing.Process(target=MqRPCClient._start_dead_msg_consumer)
            MqRPCClient.__dead_msg_consumer_process.start()
